package top.houry.netty.barrage.service.impl;

import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.google.protobuf.TextFormat;
import io.netty.channel.ChannelHandlerContext;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import top.houry.netty.barrage.annotation.BarrageAnnotation;
import top.houry.netty.barrage.bo.BarrageMsgBo;
import top.houry.netty.barrage.config.BarrageNettyConfigProperties;
import top.houry.netty.barrage.consts.BarrageKafkaConst;
import top.houry.netty.barrage.consts.BarrageMsgTypeConst;
import top.houry.netty.barrage.consts.BarrageRedisKeyConst;
import top.houry.netty.barrage.consts.BarrageVideoConst;
import top.houry.netty.barrage.entity.BarrageMsg;

import top.houry.netty.barrage.proto.BarrageProto;
import top.houry.netty.barrage.rocketmq.ProducerService;
import top.houry.netty.barrage.service.IBarrageMsgSendToClientService;
import top.houry.netty.barrage.service.IBarrageMsgService;
import top.houry.netty.barrage.service.IBarrageMsgTypeService;
import top.houry.netty.barrage.utils.BarrageConnectInfoUtils;
import top.houry.netty.barrage.utils.BarrageMsgSensitiveUtils;
import top.houry.netty.barrage.utils.BarrageRedisUtils;

import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.List;

/**
 * @Desc 服务器接收客户端发送过来的弹幕信息
 * @Author houruiyang
 * @Date 2022/1/18
 **/
@Service
@BarrageAnnotation(msgType = BarrageMsgTypeConst.WEB_CLIENT_BARRAGE_REQ)
@Slf4j
public class BarrageReceiveMsgServiceImpl implements IBarrageMsgTypeService {
    @Value("${redis.channel.msgToAll}")
    private String msgToAll;
    private IBarrageMsgSendToClientService barrageSendMsgToClientService;
    private IBarrageMsgService barrageMsgService;
//    private BarrageKafkaProducer barrageKafkaProducer;
    private ProducerService producerService;
    private BarrageNettyConfigProperties nettyConfigProperties;

    public BarrageNettyConfigProperties getNettyConfigProperties() {
        return nettyConfigProperties;
    }

    @Autowired
    public void setNettyConfigProperties(BarrageNettyConfigProperties nettyConfigProperties) {
        this.nettyConfigProperties = nettyConfigProperties;
    }
    @Autowired
    public void setProducerService(ProducerService producerService){
        this.producerService = producerService;
    }

    @Autowired
    public void setBarrageMsgService(IBarrageMsgService barrageMsgService) {
        this.barrageMsgService = barrageMsgService;
    }

    @Autowired
    public void setBarrageSendMsgToClientService(IBarrageMsgSendToClientService barrageSendMsgToClientService) {
        this.barrageSendMsgToClientService = barrageSendMsgToClientService;
    }

    @Override
    public void dealWithBarrageMessage(BarrageProto.Barrage barrage, ChannelHandlerContext ctx) {
        updateTraffic(ctx);
        try {
            // 1 组装消息
            BarrageProto.WebClientSendBarrageReq clientSendBarrage = BarrageProto.WebClientSendBarrageReq.parseFrom(barrage.getBytesData());
            log.info("Req-BarrageReceiveMsgServiceImpl-dealWithBarrageMessage-params:{}", TextFormat.printToUnicodeString(clientSendBarrage));
            String msg = StrUtil.blankToDefault(clientSendBarrage.getMsg(), "我们都是追梦人");
            String msgColor = StrUtil.blankToDefault(clientSendBarrage.getMsgColor(), "#FFFFFF");
            String userId = StrUtil.blankToDefault(clientSendBarrage.getUserId(), "");
            String videId = StrUtil.blankToDefault(clientSendBarrage.getVideoId(), "");
            int msgPosition = ObjectUtil.defaultIfNull(clientSendBarrage.getMsgPosition(), 0);
            int msgVideoTime = ObjectUtil.defaultIfNull(clientSendBarrage.getMsgVideoTime(), 1);

            // 2、进行敏感词汇过滤替换
            msg = filterSensitiveMsg(msg);
            // 3、组装数据库数据
            BarrageMsg barrageMsg = BarrageMsg
                    .builder()
                    .msgColor(msgColor)
                    .msgContent(msg)
                    .msgPosition(msgPosition)
                    .userId(Long.parseLong(userId))
                    .videoId(Long.parseLong(videId))
                    .videoTime(0 < msgVideoTime ? msgVideoTime : 1)
                    .createTime(new Date())
                    .updateTime(new Date())
                    .build();
            // 4 使用消息队列，异步解耦  存储数据
            producerService.sendMessage(new Message("db",JSONUtil.toJsonStr(barrageMsg).getBytes(StandardCharsets.UTF_8)));
            //barrageMsgService.saveBarrageMsg(barrageMsg);
            // 5、封装转发给其他的用户的消息
            InetAddress localHost = InetAddress.getLocalHost();
            BarrageMsgBo barrageMsgBo = new BarrageMsgBo(msg, msgColor, msgPosition, userId, videId,localHost.getHostAddress(),nettyConfigProperties.getServerPort());
            String jsonMessage = JSONUtil.toJsonStr(barrageMsgBo);
            // 6 记录历史总弹幕数集合 其实用一个字段即可
            BarrageRedisUtils.listPush(BarrageRedisKeyConst.BARRAGE_TOTAL_MSG_KEY + BarrageVideoConst.VIDE_ID, jsonMessage);
            // 7.1、集群拓展 redis实现   消息发布redis总线
//            BarrageRedisUtils.distributed(msgToAll, jsonMessage);  // 把自己的ip:port 作为集群部署的唯一标识放入消息
            // 7.2、 集群拓展 消息对列实现总线
            producerService.sendMessage(new Message(msgToAll,jsonMessage.getBytes(StandardCharsets.UTF_8)));
            // 8、推送给当前服务其他正在看该视屏的用户
            notifyOtherUser(videId, ctx, barrageMsgBo);

        } catch (Exception e) {
            log.error("Exception-BarrageReceiveMsgServiceImpl-dealWithBarrageMessage", e);
        }
    }

    /**
     * 弹幕消息流控
     *  避免用户恶意刷弹幕屏
     */
    private void updateTraffic(ChannelHandlerContext ctx) {
        InetSocketAddress ipSocket = (InetSocketAddress) ctx.channel().remoteAddress();
        String clientIp = ipSocket.getAddress().getHostAddress();
        BarrageRedisUtils.setKeyIncrementAndExpireAt(BarrageRedisKeyConst.BARRAGE_SERVER_FLOW_CONTROL_KEY + clientIp, 10);
    }


    private void notifyOtherUser(String videId, ChannelHandlerContext ctx, BarrageMsgBo barrageMsgBo) {
        List<ChannelHandlerContext> channelHandlerContextLists = BarrageConnectInfoUtils.getChannelHandlerContextListByVideId(videId);
        if (CollectionUtils.isEmpty(channelHandlerContextLists)) {
            return;
        }
        channelHandlerContextLists.stream()
                .filter(v -> !v.equals(ctx) && v.channel().isActive())
                .forEach(v -> barrageSendMsgToClientService.sendMsg(barrageMsgBo, v));
    }

    public void notifyOtherUser(String videId, BarrageMsgBo barrageMsgBo) {
        List<ChannelHandlerContext> channelHandlerContextLists = BarrageConnectInfoUtils.getChannelHandlerContextListByVideId(videId);
        if (CollectionUtils.isEmpty(channelHandlerContextLists)) {
            return;
        }
        channelHandlerContextLists.stream()
                .filter(v -> v.channel().isActive())
                .forEach(v -> barrageSendMsgToClientService.sendMsg(barrageMsgBo, v));
    }

    private String filterSensitiveMsg(String msg) {
        return BarrageMsgSensitiveUtils.replaceSensitiveMsg(msg);
    }

}
